package com.doit.day05

import scala.io.Source

object _01_电影案例的后三个练习 {
  def main(args: Array[String]): Unit = {

    val movie_line: List[String] = Source.fromFile("C:\\Users\\34650\\Desktop\\课程资料\\资料\\movies.txt").getLines().toList
    val rating_line: List[String] = Source.fromFile("C:\\Users\\34650\\Desktop\\课程资料\\资料\\ratings.txt").getLines().toList
    val user_line: List[String] = Source.fromFile("C:\\Users\\34650\\Desktop\\课程资料\\资料\\users.txt").getLines().toList

    /*
       ETL
     */
    val movies: Map[String, (String, String)] = movie_line.map(line => {
      val arr: Array[String] = line.split("_")
      //电影id    电影名称   电影类型
      (arr(0), (arr(1), arr(2)))
    }).toMap

    val ratings: List[(String, String, String, String)] = rating_line.map(line => {
      val arr: Array[String] = line.split("_")
      //user_id   电影id   评分   时间戳
      (arr(0), arr(1), arr(2), arr(3))
    })

    val users: Map[String, (String, String, String, String)] = user_line.map(line => {
      val arr: Array[String] = line.split("_")
      //user_id  性别   年龄  职业   邮编
      (arr(0), (arr(1), arr(2), arr(3), arr(4)))
    }).toMap


/*    xuqiu4(users,ratings)
      .toList
      .sortBy(_._1)
      .foreach(println)*/
    xuqiu5(users,ratings)


   /* xuqiu3(movies, ratings)
      .toList
      .sortBy(_._2)
      .reverse
      .take(5)
      .foreach(println)*/
  }

  //3.哪些年份的电影评分(平均分)最高，取最高的前五个年份
  def xuqiu3(movies: Map[String, (String, String)],ratings: List[(String, String, String, String)]) ={
    //获取电影名字和评分  ==》 join
    val yearAndScore: List[(String, String)] = ratings.map(tp => {
      val score: String = tp._3
      val movie_id: String = tp._2
      val movie_name: String = movies.getOrElse(movie_id, ("未知(0000)", "未知(0000)"))._1
      //根据电影名字获取到年份
      val year: String = movie_name.substring(movie_name.length - 5, movie_name.length - 1)
      (year, score)
    })
    //对年份进行分组聚合，取平均值
    val grouped: Map[String, List[(String, String)]] = yearAndScore.groupBy(tp => tp._1)
    grouped.map(tp=>{
      val value: List[(String, String)] = tp._2
      (tp._1,(value.map(_._2.toDouble).sum/value.size).formatted("%.2f").toDouble)
    })
  }

  //4.每个职业最喜欢的前三个电影id   观看次数最多的   每个职业观看次数最多的前三部电影
  def xuqiu4(users: Map[String, (String, String, String, String)],ratings: List[(String, String, String, String)]) ={
    //用户表获取用户id和职业
    val user_job: Map[String, String] = users.map(tp => {
      val user_id: String = tp._1
      val job_id: String = tp._2._3
      (user_id, job_id)
    })

    //(电影id  电影名称)  (电影id   score)  ==》 (电影名称 score)
    //获取电影id和职业id的对应关系  ==》一张表里面  (用户id和电影id)   (用户id和职业id)  ==》 (职业id，电影id)
    val job_movie: List[(String, String)] = ratings.map(tp => {
      val user_id: String = tp._1
      val movie_id: String = tp._2
      val job_id: String = user_job.getOrElse(user_id, "匪徒")
      (job_id, movie_id)
    })
    //分组聚合 求次数  取topN  排序 take  根据职业id进行分组  key 职业id   value list(职业相同的，看到的所有的电影id)
    val grouped: Map[String, List[(String, String)]] = job_movie.groupBy(_._1)

    grouped.map(tp=>{
      val job_id: String = tp._1
      //相同职业看到的所有的电影id
      val movie_ids: List[String] = tp._2.map(_._2)
      //求每个电影id观看的次数
      val res: List[String] = movie_ids.groupBy(movie_id => movie_id)
        .map(tp => (tp._1, tp._2.size))
        .toList
        .sortBy(_._2)
        .reverse
        .take(3)
        .map(_._1)
      (job_id,res)
    })
  }

  //5.年龄段在“18-24”的男性年轻人，最喜欢看哪10部电影（Id）
  def xuqiu5(users: Map[String, (String, String, String, String)],ratings: List[(String, String, String, String)])={
    //首先数据源涉及到几张表  ratings   users
    //users   user_id   年龄   性别

    val user_gender_age: Map[String, (String, String)] = users.map(tp => {
      val user_id: String = tp._1
      val gender: String = tp._2._1
      val age: String = tp._2._2
      (user_id, (gender, age))
    })

    // 字段应该怎么用   用user_id 把两张表关联起来 (user_id 电影id  性别  年龄)
    //ratings 电影id  user_id
    val user_movie: List[(String, String, String, String)] = ratings.map(tp => {
      val user_id: String = tp._1
      val movie_id: String = tp._2
      val tuple: (String, String) = user_gender_age.getOrElse(user_id, (null, null))
      (user_id,tuple._1,tuple._2, movie_id)
    })

    // 筛选过滤  性别为男的   年龄在18-24的人  (user_id 电影id)
    val tuples: List[(String, String, String, String)] = user_movie.filter(tp => {
      "M".equals(tp._2) && "18".equals(tp._3) && tp._3 != null
      //空指针本身就是在null上调用了方法
    })

    val user_movies: List[(String, String)] = tuples.map(tp => (tp._1, tp._4))

    val id_user: Map[String, Int] = user_movies.groupBy(_._2)
      .map(tp => {
        val movie_id: String = tp._1
        val value: List[(String, String)] = tp._2
        val users: List[String] = value.map(_._1).distinct
        (movie_id, users.size)
      })
    id_user.toList
      .sortBy(_._2)
      .reverse
      .take(10)
      .foreach(println)

  }



}
